package com.bawei.flink.tableapi;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DynamicTableDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //tEnv.sqlQuery("SET table.local-time-zone=Asia/Shanghai");
        tEnv.executeSql("CREATE TABLE t_clicks(\n" +
                "  user_name STRING,\n" +
                "  url STRING,\n" +
                "  ts BIGINT,\n" +
                "  time_ltz AS TO_TIMESTAMP_LTZ(ts,3),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' MINUTE\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'sink.parallelism' = '1',\n" +
                " 'topic' = 'clicks',\n" +
                " 'properties.bootstrap.servers' = 'cdh1:9092,cdh2:9092,cdh3:9092',\n" +
                " 'properties.group.id' = 'testGroup',\n" +
                " 'scan.startup.mode' = 'latest-offset',\n" +
                " 'format' = 'csv',\n" +
                " 'csv.ignore-parse-errors' = 'true',\n" +
                " 'csv.allow-comments' = 'true'\n" +
                ")");

        tEnv.executeSql("CREATE TABLE t_res(\n" +
                "  user_name STRING,\n" +
                "  window_end TIMESTAMP(3),\n" +
                "  cnts BIGINT\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'sink.parallelism' = '1',\n" +
                " 'topic' = 't_res',\n" +
                " 'properties.bootstrap.servers' = 'cdh1:9092,cdh2:9092,cdh3:9092',\n" +
                " 'format' = 'csv',\n" +
                " 'csv.ignore-parse-errors' = 'true',\n" +
                " 'csv.allow-comments' = 'true'\n" +
                ")");

        tEnv.executeSql("insert into t_res select user_name,TUMBLE_END(time_ltz,INTERVAL '1' HOUR),count(url) as cnts from t_clicks group by user_name,TUMBLE(time_ltz, INTERVAL '1' HOUR)");
    }

}
